package com.fwmagic.flink.examplecase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public class LocalKeyByFlatMap implements FlatMapFunction<String, Tuple2<String, Long>> {

    //本地 buffer，存放 local 端缓存的 app 的 pv 信息
    private HashMap<String, Long> localPvStat;

    //缓存的数据量大小，即：缓存多少数据再向下游发送
    private int batchSize;

    //计数器，获取当前批次接收的数据量
    private AtomicInteger currentSize = new AtomicInteger(0);

    LocalKeyByFlatMap(int batchSize){
        this.batchSize = batchSize;
    }

    @Override
    public void flatMap(String in, Collector collector) throws Exception {
        //  将新来的数据添加到 buffer 中
        Long pv = localPvStat.getOrDefault(in, 0L);
        localPvStat.put(in, pv + 1);

        // 如果到达设定的批次，则将 buffer 中的数据发送到下游
        if(currentSize.incrementAndGet() >= batchSize){
            // 遍历 Buffer 中数据，发送到下游
            for(Map.Entry<String, Long> appIdPv: localPvStat.entrySet()) {
                collector.collect(Tuple2.of(appIdPv.getKey(), appIdPv.getValue()));
            }
            // buffer 清空，计数器清零
            localPvStat.clear();
            currentSize.set(0);
        }
    }
}